package com.ruibang.glass.etl.utils;


import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

/**
 * @Description: RocketMQ消息的生产者
 * @Author: hwk
 */

//@Component
@Slf4j
public class RocketMqUtil {
    @Autowired
    private RocketMQTemplate rocketMqTemplate;

    public void sendMsg(String topic,String data) {
        rocketMqTemplate.convertAndSend(topic,data);
        log.info("【RocketMQ】发送同步消息：{}", data);
    }

    public void asyncSend(String topic, String tag, String data,Integer messageDelayLevel) {
        rocketMqTemplate.asyncSend(topic + ":" + tag, MessageBuilder.withPayload(data).build(), new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        // 消息发送成功
                        log.error("消息发送成功"+sendResult);
                    }

                    @Override
                    public void onException(Throwable throwable) {
                        // 消息发送异常
                        log.error("异步发送消息异常。topic：" + topic + "；tag：" + tag + "；mqMsg" + data, throwable);
                    }
                },
                3000L,
                // messageDelayLevel = 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 1d
                // messageDelayLevel = 1  2  3   4   5  6  7  8  9  10 11 12 13 14  15  16  17 18 19
                messageDelayLevel);
    }

    /**
     * 发送同步消息：消息响应后发送下一条消息
     *
     * @param topic 消息主题
     * @param tag   消息tag
     * @param key   业务号
     * @param data  消息内容
     */
    public void sendSyncMsg(String topic, String tag, String key, String data) {
        //消息
        Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
        //主题
        String destination = topic + ":" + tag;
        SendResult sendResult = rocketMqTemplate.syncSend(destination, message);

        log.info("【RocketMQ】发送同步消息：{}", sendResult);
    }

    /**
     * 发送异步消息：异步回调通知消息发送的状况
     *
     * @param topic 消息主题
     * @param tag   消息tag
     * @param key   业务号
     * @param data  消息内容
     */
    public void sendAsyncMsg(String topic, String tag, String key, String data) {
        //消息
        Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
        //主题
        String destination = topic + ":" + tag;
        rocketMqTemplate.asyncSend(destination, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("【RocketMQ】发送异步消息：{}", sendResult);
            }

            @Override
            public void onException(Throwable e) {
                log.info("【RocketMQ】发送异步消息异常：{}", e.getMessage());
            }
        });
    }


    /**
     * 发送单向消息：消息发送后无响应，可靠性差，效率高
     *
     * @param topic 消息主题
     * @param tag   消息tag
     * @param key   业务号
     * @param data  消息内容
     */
    public void sendOneWayMsg(String topic, String tag, String key, String data) {
        //消息
        Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
        //主题
        String destination = topic + ":" + tag;
        rocketMqTemplate.sendOneWay(destination, message);
    }


    /**
     * 同步延迟消息
     *
     * @param topic      主题
     * @param tag        标签
     * @param key        业务号
     * @param data       消息体
     * @param timeout    发送消息的过期时间
     * @param delayLevel 延迟等级-----固定等级：1到18分别对应1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     *
     */
    public void sendSyncDelayMsg(String topic, String tag, String key, String data, long timeout, int delayLevel) {
        // messageDelayLevel = 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 1d
        // messageDelayLevel = 1  2  3   4   5  6  7  8  9  10 11 12 13 14  15  16  17 18 19
        //消息
        Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
        //主题
        String destination = topic + ":" + tag;
        SendResult sendResult = rocketMqTemplate.syncSend(destination, message, timeout, delayLevel);
        log.info("【RocketMQ】发送同步延迟消息：{}", sendResult);
    }


    /**
     * 异步延迟消息
     *
     * @param topic      主题
     * @param tag        标签
     * @param key        业务号
     * @param data       消息体
     * @param timeout    发送消息的过期时间
     * @param delayLevel 延迟等级-----固定等级：1到18分别对应1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     */
    public void sendAsyncDelayMsg(String topic, String tag, String key, String data, long timeout, int delayLevel) {
        //消息
        Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
        //主题
        String destination = topic + ":" + tag;
        rocketMqTemplate.asyncSend(destination, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("【RocketMQ】发送异步延迟消息：{}", sendResult);
            }

            @Override
            public void onException(Throwable e) {
                log.info("【RocketMQ】发送异步延迟消息异常：{}", e.getMessage());
            }
        }, timeout, delayLevel);
    }


    /**
     * 同步顺序消息
     *
     * @param topic 主题
     * @param tag   标签
     * @param key   业务号
     * @param data  消息体
     */
    public void sendSyncOrderlyMsg(String topic, String tag, String key, String data) {
        //消息
        Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
        //主题
        String destination = topic + ":" + tag;
        SendResult sendResult = rocketMqTemplate.syncSendOrderly(destination, message, key);
        log.info("【RocketMQ】发送同步顺序消息：{}", sendResult);
    }


    /**
     * 异步顺序消息
     *
     * @param topic 主题
     * @param tag   标签
     * @param key   业务号
     * @param data  消息体
     */
    public void sendAsyncOrderlyMsg(String topic, String tag, String key, String data) {
        //消息
        Message message = MessageBuilder.withPayload(data).setHeader(RocketMQHeaders.KEYS, key).build();
        //主题
        String destination = topic + ":" + tag;
        rocketMqTemplate.asyncSendOrderly(destination, message, key, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("【RocketMQ】发送异步顺序消息：{}", sendResult);
            }
            @Override
            public void onException(Throwable e) {
                log.info("【RocketMQ】发送异步顺序消息异常：{}", e.getMessage());
            }
        });
    }
}